package com.talent.study.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Author: hszhou
 * @Date: 2025/5/9 9:27
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class TestProducer {
    private static final String TOPIC_NAME = "ws";

    public static void main(String[] args) {


        // 1. 配置生产者参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.28.33.102:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 可选优化配置
        props.put(ProducerConfig.ACKS_CONFIG, "all");           // 消息确认机制
        props.put(ProducerConfig.RETRIES_CONFIG, 3);            // 重试次数
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 幂等性

        // 2. 创建生产者实例
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {

            // 3. 构造消息记录
            ProducerRecord<String, String> record = new ProducerRecord<>(
                    TOPIC_NAME,
                    "message-key",   // 消息键（可选）
                    "Hello Kafka!"   // 消息值
            );

            // 4. 发送消息（同步/异步两种方式）

            // 同步发送（阻塞直到收到响应）
            RecordMetadata metadata = producer.send(record).get();
            System.out.printf("同步发送成功 -> topic:%s partition:%d offset:%d%n",
                    metadata.topic(), metadata.partition(), metadata.offset());

            // 异步发送（推荐生产环境使用）
//            producer.send(record, (metadata, exception) -> {
//                if (exception == null) {
//                    System.out.printf("异步发送成功 -> topic:%s partition:%d offset:%d%n",
//                            metadata.topic(), metadata.partition(), metadata.offset());
//                } else {
//                    System.err.println("发送失败: " + exception.getMessage());
//                }
//            });

        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

    }
}
